package com.edfeff.nio.http;

import com.edfeff.bio.http.StsException;
import com.edfeff.nio.ReactorSimpleHttpServer;

import java.io.IOException;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class Poller implements Runnable {

    private final ReactorSimpleHttpServer server;

    private Selector selector;
    protected BlockingQueue<Runnable> events = new LinkedBlockingQueue<>();

    public Poller(ReactorSimpleHttpServer server) {
        this.server = server;
        try {
            this.selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
            throw new StsException("启动失败");
        }
    }

    public void register(final SocketChannel socket) {
        Runnable event = new Runnable() {
            @Override
            public void run() {
                try {
                    socket.register(selector, SelectionKey.OP_READ);
                } catch (ClosedChannelException e) {
                    e.printStackTrace();
                }
            }
        };
        this.events.offer(event);
    }


    @Override
    public void run() {
        while (server.isRun()) {
            try {
                Runnable event = events.poll();
                if (event != null) {
                    event.run();
                }
                //
                int keyCount = 0;
                keyCount = selector.select(1000);
                if (keyCount == 0)
                    continue;
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey sk = iterator.next();
                    iterator.remove();
                    if (sk.isValid()) {
                        sk.interestOps(0);
                        if (sk.isReadable() || sk.isWritable()) {
                            SocketChannel channel = (SocketChannel) sk.channel();
                            server.processSocketHandler(channel);
                        }
                    } else {
                        sk.cancel();
                    }
                }

            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
}
